package com.qingcloud.ftp.test;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * test sql.
 */
public class Main2 {
	public static final String FTP_CONNECTOR_CSV = "" +
		"CREATE TABLE FTP_CONNECTOR_CSV(readName string,cellPhone string,universityName string,city string,street string,ip string)" +
		"WITH(" +
		"'connector' = 'ftp'," +
		"'host'='42.193.101.183'," +
		"'username'='ftptest'," +
		"'password'='Xfh300823@'," +
		"'path'='/u'," +
		"'format'='csv',"+
		//"'protocol'='sftp'," +
		//"'column-delimiter'=',',"+
		 "'connectPattern' = 'port'," +
		 "'readMode'='singleStream'" +
		")";

	public static final String FTP_CONNECTOR_SINK = "" +
		"CREATE TABLE FTP_CONNECTOR_SINK(id bigint,name string,score int)" +
		"WITH(" +
		"'connector' = 'ftp'," +
		"'host'='42.193.101.183'," +
		"'username'='ftptest'," +
		"'password'='123456'," +
		"'path'='/demo/sink.json'," +
		"'format'='json'" +
		")";

	public static final String FILE_SINK = "" +
		"CREATE TABLE FILE_SINK(readName string,cellPhone string,universityName string,city string,street string,ip string)"
		//+ "PARTITIONED BY(name) "
		+ "WITH("
		+ "'connector' = 'filesystem',"
		+ "'path' = 'file:///Users/gxlevi/test',"
		+ "'format' = 'csv'" +
		")";

	public static final String FILE_SOURCE = "" +
		"CREATE TABLE FILE_SOURCE(readName string,cellPhone string,universityName string,city string,street string,ip string)"
		//+ "PARTITIONED BY(name) "
		+ "WITH("
		+ "'connector' = 'filesystem',"
		+ "'path' = 'file:///Users/gxlevi/source/user1.txt',"
		+ "'format' = 'csv'" +
		")";

	public static final String FTP_USER_JSON = "" +
		"CREATE TABLE USER_INFO(readName string,cellPhone string,universityName string,city string,street string,ip string,pt AS PROCTIME())"
		+ "WITH(" +
		"'connector' = 'ftp'," +
		"'format'='csv'," +
		"'host'='127.0.0.1'," +
		"'path'='/test/`user[0-9].txt`'," +
			//"'csv.ignore-parse-errors' = 'true'," +
			"'csv.field-delimiter' = ',',"+
//		+ "'ignoreParseErrors' = 'true',"
		//+ "'connectPattern' = 'port',"
		//+ "'delete-read' = 'true',"
		"'readMode'='singleStream'" +
		")";

	public static final String SFTP_USER_JSON = "" +
		"CREATE TABLE SFTP(readName string,cellPhone string,universityName string,city string,street string,ip string,pt AS PROCTIME())"
		+ "WITH(" +
		"'connector' = 'ftp'," +
		"'host'='127.0.0.1'," +
		"'username'='gxlevi'," +
		"'password'='xcl317'," +
		"'path'='/var/ftp/gxlevi/test/`user[0-9].txt`',"
		+ "'protocol' = 'sftp'," +
		"'format'='csv',"+
		//需要putty转换
		//"'privateKeyPath' = '/Users/gxlevi/.ssh/config'," +
		//+ "'connectPattern' = 'port',"
		//+ "'delete-read' = 'true',"
		"'readMode'='singleStream'" +
		")";
	public static final String USER_SINK = "" +
		"CREATE TABLE USER_SINK(readName string,cellPhone string,universityName string,city string,street string,ip string)"
		+ "WITH(" +
		"'connector' = 'ftp'," +
		"'host'='42.193.101.183'," +
		"'username'='ftptest'," +
		"'password'='Xfh300823@'," +
		"'path'='/sink/sink.json'," +
		"'format'='json'" +
		")";

	public static void main(String[] args) {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		env.setParallelism(3);

//		env.setStateBackend(new FsStateBackend("file:///Users/gxlevi/checkpoints"));
//		env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//		env.enableCheckpointing(1000L);
//		env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
//		env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//		env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//		env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
//		env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
//		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,10000L));
//		env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.of(5, TimeUnit.MINUTES),Time.of(10,TimeUnit.MINUTES)));

		EnvironmentSettings settings = EnvironmentSettings
			.newInstance()
			.useBlinkPlanner()
			.inStreamingMode()
			.build();
		StreamTableEnvironment tEnv = StreamTableEnvironment.create(
			env,
			settings);

		tEnv.executeSql(FTP_CONNECTOR_CSV);
		tEnv.executeSql(FTP_CONNECTOR_SINK);
		tEnv.executeSql(FILE_SINK);
		tEnv.executeSql(FTP_USER_JSON);
		tEnv.executeSql(USER_SINK);
		tEnv.executeSql(FILE_SOURCE);
		tEnv.executeSql(SFTP_USER_JSON);
		long start = System.currentTimeMillis();
		//tEnv.executeSql("select * from FTP_CONNECTOR_CSV").print();
		tEnv
			.executeSql(
				"select city,count(*) from USER_INFO group by city,tumble(pt,INTERVAL '10' SECOND)")
			.print();
		//tEnv.executeSql("select city,count(*) from USER_INFO group by city").print();
		//tEnv.executeSql("select city,count(*) from SFTP group by city").print();
		//tEnv.executeSql("insert into FILE_SINK select readName,cellPhone,universityName,city,street,ip from USER_INFO");
		//tEnv.executeSql("select id, name,sum(score) from FTP_CONNECTOR_CSV group by name,id,tumble(pt,INTERVAL '10' SECOND)").print();
		//tEnv.executeSql("select id,name,sum(score) from FTP_CONNECTOR_CSV group by name,id").print();

		//tEnv.executeSql("insert into USER_SINK select readName,cellPhone,universityName,city,street,ip from FILE_SOURCE").print();

		System.out.println("===================");
		//tEnv.executeSql("insert into FILE_SINK select readName,cellPhone,universityName,city,street,ip from USER_INFO").print();
		long end = System.currentTimeMillis();
		long duration = end - start;
		System.out.println("===================");
		System.out.println(duration);
		System.out.println("===================");
		//tEnv.executeSql("insert into FILE_SINK select id,name,score from FTP_CONNECTOR_CSV").print();
	}
}
